@mxm @pvary I would appreciate your review.

There is no guarantee, that the WriteResult serialization would remain compatible between Iceberg releases. OTOH the manifest serialization should be backwards compatible. This is the reason why manifest is used to store the temporary data which is part of a checkpoint. I don't think we would like to change this.

Also writing temporary files is costly. We need to check the cost of writing more manifest files, if we decide to go down that way. We might be better off with writing more sophisticated commit logic, and generating better commit keys (maybe add specId too)

1. Fixing a potential issue in DynamicCommitter with processing multiple WriteResult for a given table.
2. Refactoring the commit logic to produce one snapshot per spec change.

It would make things easier to review if we fixed (1) and then (2), perhaps in separate commits. I need to dig my head a bit deeper into the changes to understand them properly.

@mxm How do you envision splitting these two points? The root problem is DynamicWriteResultAggregator producing multiple commit requests per table, branch, and checkpointId triplet. The DynamicCommitter is agnostic to this and commits upstream requests; it doesn't have any special handling for partition spec changes. I cleaned up DynamicCommitter to reflect a single commit expectation, which feels like it should be part of this change.

@pvary, that's a good point. I think we can achieve the same backwards compatibility guarantees without writing temporary manifest files. And instead, write data and delete files directly into a checkpoint as Avro data. Let me experiment with this and try to get a version with Avro serialisation working, which would give us backwards compatibility.

I think we can achieve the same backwards compatibility guarantees without writing temporary manifest files. And instead, write data and delete files directly into a checkpoint as Avro data. Let me experiment with this and try to get a version with Avro serialisation working, which would give us backwards compatibility.

There is an additional thing we should consider. What happens when the attributes of the Data/Delete files are changing. With the current way - storing the manifest -, we are automatically picking up changes done in the core. If we implement our own serializer/deserializer for them, we need to keep an eye out for these changes, and implement the corresponding changes in our serializers every time.

@mxm How do you envision splitting these two points? The root problem is DynamicWriteResultAggregator producing multiple commit requests per table, branch, and checkpointId triplet. The DynamicCommitter is agnostic to this and commits upstream requests; it doesn't have any special handling for partition spec changes. I cleaned up DynamicCommitter to reflect a single commit expectation, which feels like it should be part of this change.

I've left some thoughts in #14090. I also have an implementation which I'll share. We can discuss then which is the best approach to proceed.

Basically, the idea is to use as few snapshots as possible. We can combine append-only WriteResults into a single snapshot. Whenever delete files are present, we need multiple snapshots. To make this fault tolerant, we need to store an index into the list of write results and persist it as part of the snapshot summary, similarly to the Flink checkpoint id. We can then skip any previously applied WriteResults on recovery.

As for multiple partition spec per snapshot, I couldn't find that this is not permitted in Iceberg.

Whenever delete files are present, we need multiple snapshots.

@mxm Could you please clarify why we need multiple snapshots per checkpoint when delete files are present? Wouldn't Iceberg core already handle deletions in the aggregated WriteResult from multiple parallel writers per checkpoint?
The current implementation of IcebergSink commits both deletes and appends in a single snapshot using RowDelta.

@aiborodin Regular data files and delete files can both be part of the same snapshot (actually, have to be for upsert semantics). However, we have to create a table snapshot before we process WriteResults with delete files. The reason is that data and delete files are not ordered, but deletes often require an order to be applied correctly.

For example: WriteResult w1 (append-only) and WriteResult w2 (delete + append).

If we would combine w1 and w2 into a single snapshot, Iceberg will first apply the delete files and delete the relevant rows, then apply the appends from both WriteResults. If we merged both WriteResults and created a single table snapshot, any deletes matching rows appended in w1 would not get deleted. Deletes are always applied before appends. There is no order between data files and delete files, which honestly feels like a limitation of Iceberg.

The semantics are totally different when we first create a table snapshot for w1, because any deletes by w2 would be applied on top of this snapshot and before appending data via w2. That's the reason why I think we can combine WriteResults into a single snapshot, as long as we don't see delete files. As soon as we discover a delete file, we need to create a table snapshot with the so-far aggregated WriteResults.

The semantics are totally different when we first create a table snapshot for w1, because any deletes by w2 would be applied on top of this snapshot and before appending data via w2

@mxm, so are you saying we shouldn't aggregate WriteResults with delete files? If so, it will create too many commits/snapshots with high writer parallelisms, as each writer would emit a separate WriteResult.

Yes. That's why the code in the main branch is designed that way. IMHO we can only aggregate WriteResults without delete files (append-only), unless we change Iceberg core to enforce an order on how data / delete files are applied. I'd be curious to hear @pvary's thoughts on this.

IMHO we can only aggregate WriteResults without delete files (append-only), unless we change Iceberg core to enforce an order on how data / delete files are applied. I'd be curious to hear @pvary's thoughts on this.

We can only commit files for multiple checkpoints when there are only appends/data files in the checkpoint.

For Iceberg everything which is committed in a single transaction happened at the same time. So if we have equality deletes for both checkpoints, then they will be applied together.

Consider this scenario:

• R1 insert - new data file (DF1) with R1 (PK1)
• C1 commit
• R1' update - new equality delete file with PK1 (EQ1), new data file (DF2) with R1'
• C2 commit
• R1'' update - new equality delete file with PK1 (EQ2), new data file (DF3) with R1''
• C3 commit

If we merge C2 and C3 commit, then we add EQ1 and EQ2 in the same commit, and they will be applied only for DF1. They will not be applied to DF2 or DF3 as they are added in the same commit, and as a result we will have a duplication in our table. Both R1' and R1'' will be present after C3

They will not be applied to DF2 or DF3 as they are added in the same commit, and as a result we will have a duplication in our table. Both R1' and R1'' will be present after C3

Thank you for the detailed clarification on this, @pvary and @mxm. My change does not aggregate WriteResults across checkpoints. Each checkpoint would create a separate snapshot with its own delete and data files. The DynamicCommitter code in this change evidences this:

  private void commitDeltaTxn(
      Table table,
      String branch,
      NavigableMap<Long, Committer.CommitRequest<DynamicCommittable>> pendingRequests,
      CommitSummary summary,
      String newFlinkJobId,
      String operatorId) {
    for (Map.Entry<Long, CommitRequest<DynamicCommittable>> e : pendingRequests.entrySet()) {
      // We don't commit the merged result into a single transaction because for the sequential
      // transaction txn1 and txn2, the equality-delete files of txn2 are required to be applied
      // to data files from txn1. Committing the merged one will lead to the incorrect delete
      // semantic.
      WriteResult result = e.getValue().getCommittable().writeResult();

The current change would only aggregate WriteResults across multiple parallel writers per (table, branch, checkpoint) triplet, similar to the current non-dynamic IcebergSink.

Given the above, @mxm, do you still think we need to commit multiple WriteResults separately for (table, branch, checkpoint) triplet and implement the index-based solution to guarantee idempotency as you mentioned here: #14090 (comment)? If so, could you please explain why this solution is necessary?

@pvary @mxm I added the ContentFileAvroEncoder, which serialises content files using internal Iceberg Avro serialisation classes to provide backwards compatibility. This serialisation is the same as in ManifestWriter (uses InternalData), but strips out all manifest-related properties, which we don't need for content file checkpoint persistence.
There's a similarly inspired JSON-based serialiser called ContentFileParser, which is also used in Flink, but I decided to still use Avro to keep the checkpoint size minimal.

Hey folks! I've implemented the approached outlined in 1 and 2:

• Flink: Ensure DynamicCommitter idempotence in the presence of failures #14182

@aiborodin I just saw you also pushed an update. I'll take a look.

We can only commit files for multiple checkpoints when there are only appends/data files in the checkpoint.

@pvary I wasn't suggesting that we combine WriteResults from multiple Flink checkpoints. I'm suggesting to combine the append-only WriteResults in each Flink checkpoint. Currently, every WriteResult is processed separately, which creates a lot of table snapshots.

Given the above, @mxm, do you still think we need to commit multiple WriteResults separately for (table, branch, checkpoint) triplet and implement the index-based solution to guarantee idempotency as you mentioned here: #14090 (comment)? If so, could you please explain why this solution is necessary?

@aiborodin Each table / branch pair requires a separate table snapshot. While we could combine multiple Flink checkpoints during recovery, I don't think there is much benefit from doing that. Apart from recovery, every checkpoint would normally be processed independently. We wouldn't gain much from optimizing the snapshots by combining commit request from multiple checkpoints.

I'm suggesting to combine the append-only WriteResults in each Flink checkpoint.

@mxm Just combining the append-only WriteResults in each Flink checkpoint would not resolve the issue for WriteResults with delete files (see my comment in #14182 (comment)). We should combine both appends and deletes within the same checkpoint (implemented in this change), which is valid for equality delete semantics because records with the same equality fields would always be routed to the same writer, storing all unique equality delete keys in the same delete file.

While we could combine multiple Flink checkpoints during recovery, I don't think there is much benefit from doing that. Apart from recovery, every checkpoint would normally be processed independently. We wouldn't gain much from optimizing the snapshots by combining commit request from multiple checkpoints.

My suggestion is to combine both appends and deletes for the same checkpoint, which I implemented in this change.

@mxm Just combining the append-only WriteResults in each Flink checkpoint would not resolve the issue for WriteResults with delete files (see my comment in #14182 (comment)). We should combine both appends and deletes within the same checkpoint (implemented in this change), which is valid for equality delete semantics because records with the same equality fields would always be routed to the same writer, storing all unique equality delete keys in the same delete file.

I've responded in #14182 (comment). In a nutshell, my code was doing exactly what you suggest here, committing all appends / deletes from a checkpoint, but it would additionally add WriteResults from other checkpoint if those did not contain delete files. I noticed that the code was too hard to understand, so I removed this optimization.

My suggestion is to combine both appends and deletes for the same checkpoint, which I implemented in this change.

IMHO the solution you implemented goes beyond the scope of fixing the issue. I'm open to refactoring the aggregator / committer code, but I think we should add a set of minimal changes to fix the correctness issue, along side with increased test coverage. This is what I think #14182 achieves.

IMHO the solution you implemented goes beyond the scope of fixing the issue. I'm open to refactoring the aggregator / committer code, but I think we should add a set of minimal changes to fix the correctness issue, along side with increased test coverage.

@mxm I do understand the simplicity of #14182, and I am okay with merging it as a first step. Although I think we should follow up and merge this PR as well to draw a clear boundary between aggregating temporary WriteResults (DynamicWriteResultAggregator) and committing them to a table (DynamicCommitter). It would result in clean and maintainable code in the long term with a clear separation of concerns.

I added a regression test to verify there's only one commit per table/branch/checkpoint in the dynamic sink.
@pvary @mxm, Looking forward to getting this change in and closing this story.

@aiborodin: We are trying to do too many things parallel in this PR and it makes it hard to review.

Here are the separate features I see here:

1. Fixing the correctness issue
2. Adding convenience equals and hashcode implementations for BaseFile and WriteResult
3. Moving the aggregation of the commits from the committer to the Aggregator
4. Changing the state serialization mode for the Committable

Here is how I see the changes 1-by-1:

1. Fixing is a must
2. I'm afraid that the community will push back on this. You can separate out this to another PR, and see what other people think about it. I still feel that this could be a behavioral change in the core part of Iceberg, which we should not do, but I can be convinced otherwise
3. This is a very good idea, and I fully support this
4. I think the current one is not ideal, and I'm happy to talk about it, but I suggest that we do it in a separate PR

Thank you for reporting the issue and working on the solution!

@pvary thanks for summarising and highlighting the features. I will work on extracting them into separate PRs.

@pvary I raised #14312, which addresses your point 3. Could you please take a look?

It moves commit aggregation to DynamicWriteResultAggregator and uses the current Iceberg manifest reader and writer for serialisation. As a next step, I want to improve the serialisation mechanism and encapsulate it completely into DynamicCommittableSerializer.

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the [email protected] list. Thank you for your contributions.

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.